#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date    : 2022/7/1 9:36
# @Author  : 招财大龙猫
# @File    : ftp2ftp.py
# @Description : 建立ftp到ftp 大量小文件的同步
import argparse
import datetime
import ftplib
import os
import time
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor
from math import floor

import pandas as pd

import ftputil
from ftputil.session import session_factory

# process_no = (os.cpu_count() - 1) or 1
process_no = 3  # 搬运任务调度分配到多少的进程进行搬运
threading_no = 16  # 单个进程最多同时执行ftp文件搬运的梳理
ftp_cache_max_size = 10000000  # ftp目录中文件数量级

file2file = []


def str2bool(v):
    if isinstance(v, bool):
        return v
    if v.lower() in ('yes', 'true', 't', 'y', '1'):
        return True
    elif v.lower() in ('no', 'false', 'f', 'n', '0'):
        return False
    else:
        raise argparse.ArgumentTypeError('Boolean value expected.')


parser = argparse.ArgumentParser(description='ftp2ftp 参数详细说明')
parser.add_argument('--src-ip', '-sip', help='资源ftp服务器地址', required=True)
parser.add_argument('--src-port', '-spt', type=int, help='资源ftp服务器端口，非必要参数，默认"21"', default='21')
parser.add_argument('--src-user', '-sur', help='资源ftp服务器用户，必要参数', required=True)
parser.add_argument('--src-passwd', '-spd', help='资源ftp 密码，必要参数', required=True)
parser.add_argument('--src-dir', '-sd', help='资源ftp服务器 文件夹，必要参数', required=True)
parser.add_argument('--dst-ip', '-dip', help='目的ftp服务器地址', required=True)
parser.add_argument('--dst-port', '-dpt', type=int, help='目的ftp服务器端口，非必要参数，默认"21"', default='21')
parser.add_argument('--dst-user', '-dur', help='目的ftp服务器用户，必要参数', required=True)
parser.add_argument('--dst-passwd', '-dpd', help='目的ftp 密码，必要参数', required=True)
parser.add_argument('--dst-dir', '-dd', help='目的ftp 文件夹，必要参数', required=True)
parser.add_argument('--is-delete', '-id', type=str2bool, help='源文件同步完成是否需要删除，True 删除，False 不删除，默认不删除', default=False,
                    required=True)
parser.add_argument('--is-cron', '-ic', type=str2bool, help='是否是定时任务 True 定时执行，False 实时执行，默认实时', default=False,
                    required=True)
args = parser.parse_args()


def partition(num_list, size):
    """
    Returns a new list with elements
    of which is a list of certain size.

        >>> partition([1, 2, 3, 4], 3)
        [[1, 2, 3], [4]]
    """
    return [num_list[i:i + size] for i in range(0, len(num_list), size)]


class Myftp(object):
    def __init__(self, hostaddr, username, password, dir_file, port=21):
        """
        :param hostaddr: ftp地址
        :param username: ftp账号
        :param password: ftp密码
        :param dir_file: 需要同步的ftp文件 源文件--->目的文件
        :param port:     ftp端口
        """
        self.hostaddr = hostaddr
        self.username = username
        self.password = password
        self.port = port
        self.dir_file = dir_file
        self.ftp = ftplib.FTP()
        self.ftp.set_debuglevel(0)  # 2 调试模式 ，0 正常模式
        self.ftp_util = None
        # False:主动模式   True：被动模式
        self.pasv = True

    def __ftp_del__(self):
        """
        终端ftp连接
        :return:
        """
        try:
            self.ftp.quit()
        except Exception as exp:
            self.ftp.close()

    def __ftputil_del__(self):
        self.ftp_util.close()

    def ftp_login(self):
        ftp = self.ftp
        try:
            ftp.set_pasv(self.pasv)
            ftp.connect(self.hostaddr, self.port)
            ftp.login(self.username, self.password)
            # ftp.encoding = 'GB18030'
            ftp.encoding = 'UTF-8'
        except Exception as e:
            print(f"ftp登录失败:{self.hostaddr}---->错误信息{e}")
            print(f'20s 后重新登录ftp {self.hostaddr}')
            time.sleep(20)
            self.ftp_login()

    def ftputil_login(self):
        try:
            self.ftp_util = ftputil.FTPHost(self.hostaddr, self.username, self.password,
                                            session_factory=session_factory(base_class=ftplib.FTP,
                                                                            port=self.port,
                                                                            use_passive_mode=self.pasv,
                                                                            encrypt_data_channel=True,
                                                                            encoding='UTF-8',
                                                                            debug_level=None))
            self.ftp_util.use_list_a_option = True
        except Exception as e:
            print(f"ftputil登录失败:{self.hostaddr}---->错误信息{e}")
            print(f'20s 后重新登录ftp {self.hostaddr}')
            time.sleep(20)
            self.ftputil_login()

    def get_total_size(self, file2file):
        """
        判断文件是否空
        :return: dircnt 文件夹数量  filecnt 文件数量
        """
        pathfile = pd.DataFrame()
        start_time = datetime.datetime.now()
        for parent, dirnames, filenames in self.ftp_util.walk(self.dir_file):
            if filenames:
                df = pd.DataFrame(filenames)
                if parent == '/':
                    string = parent
                else:
                    string = parent + '/'
                df[0] = string + df[0]
                # 获取文件最新修改时间
                df['st_mtime'] = df[0].apply(lambda x: self.ftp_util.lstat(x).st_mtime)
                pathfile = pd.concat([pathfile, df], axis=0)
            else:
                pass
            print(f"{parent}目录文件遍历完成")

        print('遍历文件花费时间', datetime.datetime.now() - start_time)

        if file2file and len(pathfile):
            # 去除已经搬过的文件
            df2 = pathfile
            df2 = df2[~df2[0].isin(file2file)]
        else:
            df2 = pathfile

        if len(df2):
            df2 = df2.sort_values(by='st_mtime', ascending=False)  # 同步文件最新的数据
            res = df2[0].values.tolist()
        else:
            res = []

        print(f'遍历文件的总量 {len(res)}')
        return res

    def get_file2memory(self, finsh_list):
        """
        将远端文件映射到缓存中
        :return: 缓存中的文件
        """
        memory_file = None
        ftp_host = self.ftp_util
        # 关闭ftp_host缓存机制
        # ftp_host.stat_cache.disable()
        ftp_host.stat_cache.resize(ftp_cache_max_size)  # 设置缓存大小
        ftp_host.stat_cache.max_age = 60 * 60  # 缓存中条目的最长期限设置为一小时
        ftp_host.keep_alive()  # 保证连接活跃
        if ftp_host.path.isfile(self.dir_file):
            # 如果是一个文件
            print(f"{self.hostaddr},ftp 只支持文件夹同步，不支持单个文件同步")
        elif ftp_host.path.isdir(self.dir_file):
            # 如果是一个文件夹
            memory_file = self.get_total_size(finsh_list)
            if not memory_file:
                print(f'{self.hostaddr},ftp的目录{self.dir_file}为空')
            else:
                memory_file = memory_file
        return memory_file


def process_item(items,
                 src_ip, src_port, src_user, src_passwd, src_dir,
                 dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                 is_delete
                 ):
    # 转线程
    mid_file2file = []
    flen = 0
    threading_pool = ThreadPoolExecutor(threading_no)
    objs = []
    for file in items:
        if dst_dir == '/':
            dst_file = file.replace(src_dir + '/', dst_dir, 1)
        else:
            dst_file = file.replace(src_dir, dst_dir + '/', 1)
        future = threading_pool.submit(ftptoftp, src_ip, src_port, src_user, src_passwd, src_dir,
                                       dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                                       file, dst_file, is_delete)
        objs.append(future)

    threading_pool.shutdown(wait=True)

    for obj in objs:
        filesize, filename = obj.result()
        flen = flen + int(filesize)
        if filename:
            mid_file2file.append(filename)

    return flen, mid_file2file


def ftptoftp(src_ip, src_port, src_user, src_passwd, src_dir,
             dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
             local_dir, remote_dir, delete=False):
    src_ftp_item = Myftp(src_ip, src_user, src_passwd, src_dir, src_port)
    src_ftp_item.ftp_login()
    dst_ftp_item = Myftp(dst_ip, dst_user, dst_passwd, dst_dir, dst_port)
    dst_ftp_item.ftp_login()

    # 分割目的端目录，文件
    mid_dst_dir = remote_dir.split('/')
    mid_dst_dir.pop(-1)
    if len(mid_dst_dir) == 1:
        mid_dst_dir = '/'
    else:
        mid_dst_dir = '/'.join(mid_dst_dir)

    # 判断文件目录是否存在
    try:
        dst_ftp_item.ftp.mkd(mid_dst_dir)
    except:
        pass

    done_file = False
    flen = 0
    try:
        src_ftp_item.ftp.voidcmd('TYPE I')
        dst_ftp_item.ftp.voidcmd('TYPE I')
        sock1 = src_ftp_item.ftp.transfercmd(f"RETR {local_dir}")
        sock2 = dst_ftp_item.ftp.transfercmd(f"STOR {remote_dir}")
        # print(f'********{local_dir}***********')
        while True:
            block = sock1.recv(1024)

            if len(block) == 0:
                break
            flen += len(block)

            while len(block) > 0:
                sentlen = sock2.send(block)
                block = block[sentlen:]

        sock1.close()
        sock2.close()
    except Exception as exp:
        print(f'{local_dir}上传失败,错误信息:', exp)
    else:
        if delete:  # 是否删除源文件
            try:
                src_list = local_dir.split('/')
                src_filename = src_list.pop(-1)
                if len(src_list) == 1:
                    mid_src_dir = '/'
                else:
                    mid_src_dir = '/'.join(src_list)
                src_ftp_item.ftp.cwd(mid_src_dir)
                try:
                    src_ftp_item.ftp.delete(src_filename)
                except:
                    pass
            except Exception as exp:
                print(f'删除文件{local_dir}失败，失败信息为{exp}')
        else:
            done_file = local_dir  # 添加已经上传的文件记录
            # share_list.append(local_dir)

        src_ftp_item.ftp.getresp()
        dst_ftp_item.ftp.getresp()

        if not flen:
            print(f"传输文件 {local_dir} 无内容！")
        else:
            print(f"传输文件 {local_dir}------>{remote_dir}, 文件大小: {flen} bytes")
    finally:
        src_ftp_item.__ftp_del__()
        dst_ftp_item.__ftp_del__()

    return flen, done_file


def real_time_main(src_ip, src_port, src_user, src_passwd, src_dir,
                   dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                   is_delete):
    """
    实时执行的任务
    :param src_ip: 资源端 ip
    :param src_port: 资源端 端口
    :param src_user: 资源端 用户
    :param src_passwd: 资源端 密码
    :param src_dir: 资源端 文件夹
    :param dst_ip: 目的端 ip
    :param dst_port: 目的端 端口
    :param dst_user:  目的端 用户
    :param dst_dir:  目的端 文件夹
    :param is_delete: 资源端文件是否删除  True 删除 False 不删除
    :return:
    """
    global file2file
    while True:
        try:
            src_ftp = Myftp(src_ip, src_user, src_passwd, src_dir, src_port)
            src_ftp.ftputil_login()
            memory_file = src_ftp.get_file2memory(file2file)
            src_ftp.__ftputil_del__()
            if memory_file:
                flen = 0
                strat_time = datetime.datetime.now()
                if (int(floor(len(memory_file) / process_no)) < 1) or (len(memory_file) < 200):
                    for file in memory_file:
                        if dst_dir == '/':
                            dst_file = file.replace(src_dir + '/', dst_dir, 1)
                        else:
                            dst_file = file.replace(src_dir, dst_dir + '/', 1)
                        res_flen, done_file = ftptoftp(src_ip, src_port, src_user, src_passwd, src_dir,
                                                       dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                                                       file, dst_file,
                                                       is_delete)
                        flen = flen + int(res_flen)
                        if done_file:
                            file2file.append(done_file)
                else:
                    futures = []
                    rets_files = partition(memory_file, int(floor(len(memory_file) / process_no)))
                    process_pool = ProcessPoolExecutor(process_no)
                    for rets_file in rets_files:
                        future = process_pool.submit(process_item, rets_file,
                                                     src_ip, src_port, src_user, src_passwd, src_dir,
                                                     dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                                                     is_delete)
                        futures.append(future)

                    process_pool.shutdown(wait=True)

                    for obj in futures:
                        filesize, filename_ls = obj.result()
                        flen = flen + int(filesize)
                        file2file = file2file + filename_ls

                end_time = datetime.datetime.now()
                task_time = (end_time - strat_time).total_seconds()
                print(
                    f"同步源文件 {src_ftp.dir_file} 完成,文件总大小: {flen} bytes, 平均网速 {flen / task_time} bytes/s，花费时间 {task_time} s")
            else:
                print(f"同步源文件 {src_ftp.dir_file} 完成, 文件大小: 0 bytes, 平均网速 0 bytes/s，花费时间 0 s")
            print('搬运文件缓存数量', len(file2file))
            if len(file2file) >= 100000:
                file2file = file2file[-100000:]  # 保存10W的记录
        except Exception as exp:
            print('ftp, 实时文件同步异常', exp)
            time.sleep(20)
            print('重新执行ftp同步')
            real_time_main(src_ip, src_port, src_user, src_passwd, src_dir,
                           dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                           is_delete)
        finally:
            time.sleep(2)


def crontab_time_main(src_ip, src_port, src_user, src_passwd, src_dir,
                      dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                      is_delete):
    """
    实时执行的任务
    :param src_ip: 资源端 ip
    :param src_port: 资源端 端口
    :param src_user: 资源端 用户
    :param src_passwd: 资源端 密码
    :param src_dir: 资源端 文件夹
    :param dst_ip: 目的端 ip
    :param dst_port: 目的端 端口
    :param dst_user:  目的端 用户
    :param dst_dir:  目的端 文件夹
    :param is_delete: 资源端文件是否删除  True 删除 False 不删除
    :return:
    """
    global file2file
    try:
        src_ftp = Myftp(src_ip, src_user, src_passwd, src_dir, src_port)
        src_ftp.ftputil_login()
        memory_file = src_ftp.get_file2memory(file2file)
        src_ftp.__ftputil_del__()
        if memory_file:
            flen = 0
            strat_time = datetime.datetime.now()
            if (int(floor(len(memory_file) / process_no)) < 1) or (len(memory_file) < 200):
                for file in memory_file:
                    if dst_dir == '/':
                        dst_file = file.replace(src_dir + '/', dst_dir, 1)
                    else:
                        dst_file = file.replace(src_dir, dst_dir + '/', 1)
                    res_flen, done_file = ftptoftp(src_ip, src_port, src_user, src_passwd, src_dir,
                                                   dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                                                   file, dst_file,
                                                   is_delete)
                    flen = flen + int(res_flen)
                    if done_file:
                        file2file.append(done_file)
            else:
                futures = []
                rets_files = partition(memory_file, int(floor(len(memory_file) / process_no)))
                process_pool = ProcessPoolExecutor(process_no)
                for rets_file in rets_files:
                    future = process_pool.submit(process_item, rets_file,
                                                 src_ip, src_port, src_user, src_passwd, src_dir,
                                                 dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                                                 is_delete)
                    futures.append(future)

                process_pool.shutdown(wait=True)

                for obj in futures:
                    filesize, filename_ls = obj.result()
                    flen = flen + int(filesize)
                    file2file = file2file + filename_ls

            end_time = datetime.datetime.now()
            task_time = (end_time - strat_time).total_seconds()
            print(
                f"同步源文件 {src_ftp.dir_file} 完成,文件总大小: {flen} bytes, 平均网速 {flen / task_time} bytes/s，花费时间 {task_time} s")
        else:
            print(f"同步源文件 {src_ftp.dir_file} 完成, 文件大小: 0 bytes, 平均网速 0 bytes/s，花费时间 0 s")
        print('搬运文件缓存数量', len(file2file))
    except Exception as exp:
        print('ftp, 定时文件同步异常', exp)


if __name__ == '__main__':
    # '192.168.13.236', 'ftpadmin', 'wuhan2020', '/', 21
    # '192.168.13.232', 'ftp_user', '123456', '/', 21
    # '172.16.2.237', 'admin', '1', '/', 21
    # '172.16.2.144', 'admin', '1', '/', 21
    # 传入的源端信息
    print(args.__dict__)
    src_ip = args.src_ip
    src_port = args.src_port
    src_user = args.src_user
    src_passwd = args.src_passwd
    src_dir = args.src_dir

    # 传入的目的端信息
    dst_ip = args.dst_ip
    dst_port = args.dst_port
    dst_user = args.dst_user
    dst_passwd = args.dst_passwd
    dst_dir = args.dst_dir

    # 传入的控制信息
    is_delete = args.is_delete
    is_cron = args.is_cron
    # is_cron = False
    if is_cron:
        # 定时任务
        crontab_time_main(src_ip, src_port, src_user, src_passwd, src_dir,
                          dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                          is_delete)
    else:
        # 实时任务
        # real_time_main('192.168.13.236', 21, 'ftpadmin', 'wuhan2020', '/test',
        #                '172.16.2.237', 21, 'admin', '1', '/test3',
        #                False)
        real_time_main(src_ip, src_port, src_user, src_passwd, src_dir,
                       dst_ip, dst_port, dst_user, dst_passwd, dst_dir,
                       is_delete)
